--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
sbapp/pmqtt/subscribe.py 1.9.2 (761c4329) Text, 11.67 KB
T8b949e# Copyright (c) 2016 Roger Light <roger@atchoo.org>
T8b949e#
T8b949e# All rights reserved. This program and the accompanying materials
T8b949e# are made available under the terms of the Eclipse Public License v2.0
T8b949e# and Eclipse Distribution License v1.0 which accompany this distribution.
T8b949e#
T8b949e# The Eclipse Public License is available at
T8b949e# http://www.eclipse.org/legal/epl-v20.html
T8b949e# and the Eclipse Distribution License is available at
T8b949e# http://www.eclipse.org/org/documents/edl-v10.php.
T8b949e#
T8b949e# Contributors:
T8b949e# Roger Light - initial API and implementation
T8b949e"""
This module provides some helper functions to allow straightforward subscribing
to topics and retrieving messages. The two functions are simple(), which
returns one or messages matching a set of topics, and callback() which allows
you to pass a callback for processing of messages.
"""
Tff7b72from T7ee787.T7ee787. Tff7b72import Te6edf3mqtt
Tff7b72from T7ee787. Tff7b72import Te6edf3client Tff7b72as Te6edf3paho
Tff7b72def Td2a8ff_on_connectTb4b4b4(Te6edf3clientTb4b4b4, Te6edf3userdataTb4b4b4, Te6edf3flagsTb4b4b4, Te6edf3reason_codeTb4b4b4, Te6edf3propertiesTb4b4b4)Tb4b4b4:
T8b949e"""Internal callback"""
Tff7b72if Te6edf3reason_code Tff7b72!= T79c0ff0Tb4b4b4:
Tff7b72raise Te6edf3mqttTff7b72.Td2a8ffMQTTExceptionTb4b4b4(Te6edf3pahoTff7b72.Td2a8ffconnack_stringTb4b4b4(Te6edf3reason_codeTb4b4b4)Tb4b4b4)
Tff7b72if Tffa657isinstanceTb4b4b4(Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6fftopicsTa5d6ff'Tb4b4b4]Tb4b4b4, Tffa657listTb4b4b4)Tb4b4b4:
Tff7b72for Te6edf3topic Tff7b72in Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6fftopicsTa5d6ff'Tb4b4b4]Tb4b4b4:
Te6edf3clientTff7b72.Td2a8ffsubscribeTb4b4b4(Te6edf3topicTb4b4b4, Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffqosTa5d6ff'Tb4b4b4]Tb4b4b4)
Tff7b72elseTb4b4b4:
Te6edf3clientTff7b72.Td2a8ffsubscribeTb4b4b4(Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6fftopicsTa5d6ff'Tb4b4b4]Tb4b4b4, Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffqosTa5d6ff'Tb4b4b4]Tb4b4b4)
Tff7b72def Td2a8ff_on_message_callbackTb4b4b4(Te6edf3clientTb4b4b4, Te6edf3userdataTb4b4b4, Te6edf3messageTb4b4b4)Tb4b4b4:
T8b949e"""Internal callback"""
Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffcallbackTa5d6ff'Tb4b4b4]Tb4b4b4(Te6edf3clientTb4b4b4, Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffuserdataTa5d6ff'Tb4b4b4]Tb4b4b4, Te6edf3messageTb4b4b4)
Tff7b72def Td2a8ff_on_message_simpleTb4b4b4(Te6edf3clientTb4b4b4, Te6edf3userdataTb4b4b4, Te6edf3messageTb4b4b4)Tb4b4b4:
T8b949e"""Internal callback"""
Tff7b72if Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmsg_countTa5d6ff'Tb4b4b4] Tff7b72== T79c0ff0Tb4b4b4:
Tff7b72return
T8b949e# Don't process stale retained messages if 'retained' was false
Tff7b72if Te6edf3messageTff7b72.Td2a8ffretain Tff7b72and Tff7b72not Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffretainedTa5d6ff'Tb4b4b4]Tb4b4b4:
Tff7b72return
Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmsg_countTa5d6ff'Tb4b4b4] Tff7b72= Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmsg_countTa5d6ff'Tb4b4b4] Tff7b72- T79c0ff1
Tff7b72if Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmessagesTa5d6ff'Tb4b4b4] Tff7b72is Tff7b72None Tff7b72and Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmsg_countTa5d6ff'Tb4b4b4] Tff7b72== T79c0ff0Tb4b4b4:
Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmessagesTa5d6ff'Tb4b4b4] Tff7b72= Te6edf3message
Te6edf3clientTff7b72.Td2a8ffdisconnectTb4b4b4(Tb4b4b4)
Tff7b72return
Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmessagesTa5d6ff'Tb4b4b4]Tff7b72.Td2a8ffappendTb4b4b4(Te6edf3messageTb4b4b4)
Tff7b72if Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmsg_countTa5d6ff'Tb4b4b4] Tff7b72== T79c0ff0Tb4b4b4:
Te6edf3clientTff7b72.Td2a8ffdisconnectTb4b4b4(Tb4b4b4)
Tff7b72def Td2a8ffcallbackTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3topicsTb4b4b4, Te6edf3qosTff7b72=T79c0ff0Tb4b4b4, Te6edf3userdataTff7b72=Tff7b72NoneTb4b4b4, Te6edf3hostnameTff7b72=Ta5d6ff"Ta5d6fflocalhostTa5d6ff"Tb4b4b4,
Te6edf3portTff7b72=T79c0ff1883Tb4b4b4, Te6edf3client_idTff7b72=Ta5d6ff"Ta5d6ff"Tb4b4b4, Te6edf3keepaliveTff7b72=T79c0ff60Tb4b4b4, Te6edf3willTff7b72=Tff7b72NoneTb4b4b4, Te6edf3authTff7b72=Tff7b72NoneTb4b4b4,
Te6edf3tlsTff7b72=Tff7b72NoneTb4b4b4, Te6edf3protocolTff7b72=Te6edf3pahoTff7b72.Td2a8ffMQTTv311Tb4b4b4, Te6edf3transportTff7b72=Ta5d6ff"Ta5d6fftcpTa5d6ff"Tb4b4b4,
Te6edf3clean_sessionTff7b72=Tff7b72TrueTb4b4b4, Te6edf3proxy_argsTff7b72=Tff7b72NoneTb4b4b4)Tb4b4b4:
T8b949e"""Subscribe to a list of topics and process them in a callback function.
This function creates an MQTT client, connects to a broker and subscribes
to a list of topics. Incoming messages are processed by the user provided
callback. This is a blocking function and will never return.
:param callback: function with the same signature as `on_message` for
processing the messages received.
:param topics: either a string containing a single topic to subscribe to, or a
list of topics to subscribe to.
:param int qos: the qos to use when subscribing. This is applied to all topics.
:param userdata: passed to the callback
:param str hostname: the address of the broker to connect to.
Defaults to localhost.
:param int port: the port to connect to the broker on. Defaults to 1883.
:param str client_id: the MQTT client id to use. If "" or None, the Paho library will
generate a client id automatically.
:param int keepalive: the keepalive timeout value for the client. Defaults to 60
seconds.
:param will: a dict containing will parameters for the client: will = {'topic':
"<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
Topic is required, all other parameters are optional and will
default to None, 0 and False respectively.
Defaults to None, which indicates no will should be used.
:param auth: a dict containing authentication parameters for the client:
auth = {'username':"<username>", 'password':"<password>"}
Username is required, password is optional and will default to None
if not provided.
Defaults to None, which indicates no authentication is to be used.
:param tls: a dict containing TLS configuration parameters for the client:
dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
'ciphers':"<ciphers">, 'insecure':"<bool>"}
ca_certs is required, all other parameters are optional and will
default to None if not provided, which results in the client using
the default behaviour - see the paho.mqtt.client documentation.
Alternatively, tls input can be an SSLContext object, which will be
processed using the tls_set_context method.
Defaults to None, which indicates that TLS should not be used.
:param str transport: set to "tcp" to use the default setting of transport which is
raw TCP. Set to "websockets" to use WebSockets as the transport.
:param clean_session: a boolean that determines the client type. If True,
the broker will remove all information about this client
when it disconnects. If False, the client is a persistent
client and subscription information and queued messages
will be retained when the client disconnects.
Defaults to True.
:param proxy_args: a dictionary that will be given to the client.
"""
Tff7b72if Te6edf3qos Tff7b72< T79c0ff0 Tff7b72or Te6edf3qos Tff7b72> T79c0ff2Tb4b4b4:
Tff7b72raise Tf85149ValueErrorTb4b4b4(Ta5d6ff'Ta5d6ffqos must be in the range 0-2Ta5d6ff'Tb4b4b4)
Te6edf3callback_userdata Tff7b72= Tb4b4b4{
Ta5d6ff'Ta5d6ffcallbackTa5d6ff'Tb4b4b4:Te6edf3callbackTb4b4b4,
Ta5d6ff'Ta5d6fftopicsTa5d6ff'Tb4b4b4:Te6edf3topicsTb4b4b4,
Ta5d6ff'Ta5d6ffqosTa5d6ff'Tb4b4b4:Te6edf3qosTb4b4b4,
Ta5d6ff'Ta5d6ffuserdataTa5d6ff'Tb4b4b4:Te6edf3userdataTb4b4b4}
Te6edf3client Tff7b72= Te6edf3pahoTff7b72.Td2a8ffClientTb4b4b4(
Te6edf3pahoTff7b72.Td2a8ffCallbackAPIVersionTff7b72.Td2a8ffVERSION2Tb4b4b4,
Te6edf3client_idTff7b72=Te6edf3client_idTb4b4b4,
Te6edf3userdataTff7b72=Te6edf3callback_userdataTb4b4b4,
Te6edf3protocolTff7b72=Te6edf3protocolTb4b4b4,
Te6edf3transportTff7b72=Te6edf3transportTb4b4b4,
Te6edf3clean_sessionTff7b72=Te6edf3clean_sessionTb4b4b4,
Tb4b4b4)
Te6edf3clientTff7b72.Td2a8ffenable_loggerTb4b4b4(Tb4b4b4)
Te6edf3clientTff7b72.Td2a8ffon_message Tff7b72= Te6edf3_on_message_callback
Te6edf3clientTff7b72.Td2a8ffon_connect Tff7b72= Te6edf3_on_connect
Tff7b72if Te6edf3proxy_args Tff7b72is Tff7b72not Tff7b72NoneTb4b4b4:
Te6edf3clientTff7b72.Td2a8ffproxy_setTb4b4b4(Tff7b72*Tff7b72*Te6edf3proxy_argsTb4b4b4)
Tff7b72if Te6edf3authTb4b4b4:
Te6edf3username Tff7b72= Te6edf3authTff7b72.Td2a8ffgetTb4b4b4(Ta5d6ff'Ta5d6ffusernameTa5d6ff'Tb4b4b4)
Tff7b72if Te6edf3usernameTb4b4b4:
Te6edf3password Tff7b72= Te6edf3authTff7b72.Td2a8ffgetTb4b4b4(Ta5d6ff'Ta5d6ffpasswordTa5d6ff'Tb4b4b4)
Te6edf3clientTff7b72.Td2a8ffusername_pw_setTb4b4b4(Te6edf3usernameTb4b4b4, Te6edf3passwordTb4b4b4)
Tff7b72elseTb4b4b4:
Tff7b72raise Tf85149KeyErrorTb4b4b4(Ta5d6ff"Ta5d6ffThe Ta5d6ff'Ta5d6ffusernameTa5d6ff'Ta5d6ff key was not found, this is Ta5d6ff"
Ta5d6ff"Ta5d6ffrequired for authTa5d6ff"Tb4b4b4)
Tff7b72if Te6edf3will Tff7b72is Tff7b72not Tff7b72NoneTb4b4b4:
Te6edf3clientTff7b72.Td2a8ffwill_setTb4b4b4(Tff7b72*Tff7b72*Te6edf3willTb4b4b4)
Tff7b72if Te6edf3tls Tff7b72is Tff7b72not Tff7b72NoneTb4b4b4:
Tff7b72if Tffa657isinstanceTb4b4b4(Te6edf3tlsTb4b4b4, Tffa657dictTb4b4b4)Tb4b4b4:
Te6edf3insecure Tff7b72= Te6edf3tlsTff7b72.Td2a8ffpopTb4b4b4(Ta5d6ff'Ta5d6ffinsecureTa5d6ff'Tb4b4b4, Tff7b72FalseTb4b4b4)
Te6edf3clientTff7b72.Td2a8fftls_setTb4b4b4(Tff7b72*Tff7b72*Te6edf3tlsTb4b4b4)
Tff7b72if Te6edf3insecureTb4b4b4:
T8b949e# Must be set *after* the `client.tls_set()` call since it sets
T8b949e# up the SSL context that `client.tls_insecure_set` alters.
Te6edf3clientTff7b72.Td2a8fftls_insecure_setTb4b4b4(Te6edf3insecureTb4b4b4)
Tff7b72elseTb4b4b4:
T8b949e# Assume input is SSLContext object
Te6edf3clientTff7b72.Td2a8fftls_set_contextTb4b4b4(Te6edf3tlsTb4b4b4)
Te6edf3clientTff7b72.Td2a8ffconnectTb4b4b4(Te6edf3hostnameTb4b4b4, Te6edf3portTb4b4b4, Te6edf3keepaliveTb4b4b4)
Te6edf3clientTff7b72.Td2a8ffloop_foreverTb4b4b4(Tb4b4b4)
Tff7b72def Td2a8ffsimpleTb4b4b4(Te6edf3topicsTb4b4b4, Te6edf3qosTff7b72=T79c0ff0Tb4b4b4, Te6edf3msg_countTff7b72=T79c0ff1Tb4b4b4, Te6edf3retainedTff7b72=Tff7b72TrueTb4b4b4, Te6edf3hostnameTff7b72=Ta5d6ff"Ta5d6fflocalhostTa5d6ff"Tb4b4b4,
Te6edf3portTff7b72=T79c0ff1883Tb4b4b4, Te6edf3client_idTff7b72=Ta5d6ff"Ta5d6ff"Tb4b4b4, Te6edf3keepaliveTff7b72=T79c0ff60Tb4b4b4, Te6edf3willTff7b72=Tff7b72NoneTb4b4b4, Te6edf3authTff7b72=Tff7b72NoneTb4b4b4,
Te6edf3tlsTff7b72=Tff7b72NoneTb4b4b4, Te6edf3protocolTff7b72=Te6edf3pahoTff7b72.Td2a8ffMQTTv311Tb4b4b4, Te6edf3transportTff7b72=Ta5d6ff"Ta5d6fftcpTa5d6ff"Tb4b4b4,
Te6edf3clean_sessionTff7b72=Tff7b72TrueTb4b4b4, Te6edf3proxy_argsTff7b72=Tff7b72NoneTb4b4b4)Tb4b4b4:
T8b949e"""Subscribe to a list of topics and return msg_count messages.
This function creates an MQTT client, connects to a broker and subscribes
to a list of topics. Once "msg_count" messages have been received, it
disconnects cleanly from the broker and returns the messages.
:param topics: either a string containing a single topic to subscribe to, or a
list of topics to subscribe to.
:param int qos: the qos to use when subscribing. This is applied to all topics.
:param int msg_count: the number of messages to retrieve from the broker.
if msg_count == 1 then a single MQTTMessage will be returned.
if msg_count > 1 then a list of MQTTMessages will be returned.
:param bool retained: If set to True, retained messages will be processed the same as
non-retained messages. If set to False, retained messages will
be ignored. This means that with retained=False and msg_count=1,
the function will return the first message received that does
not have the retained flag set.
:param str hostname: the address of the broker to connect to.
Defaults to localhost.
:param int port: the port to connect to the broker on. Defaults to 1883.
:param str client_id: the MQTT client id to use. If "" or None, the Paho library will
generate a client id automatically.
:param int keepalive: the keepalive timeout value for the client. Defaults to 60
seconds.
:param will: a dict containing will parameters for the client: will = {'topic':
"<topic>", 'payload':"<payload">, 'qos':<qos>, 'retain':<retain>}.
Topic is required, all other parameters are optional and will
default to None, 0 and False respectively.
Defaults to None, which indicates no will should be used.
:param auth: a dict containing authentication parameters for the client:
auth = {'username':"<username>", 'password':"<password>"}
Username is required, password is optional and will default to None
if not provided.
Defaults to None, which indicates no authentication is to be used.
:param tls: a dict containing TLS configuration parameters for the client:
dict = {'ca_certs':"<ca_certs>", 'certfile':"<certfile>",
'keyfile':"<keyfile>", 'tls_version':"<tls_version>",
'ciphers':"<ciphers">, 'insecure':"<bool>"}
ca_certs is required, all other parameters are optional and will
default to None if not provided, which results in the client using
the default behaviour - see the paho.mqtt.client documentation.
Alternatively, tls input can be an SSLContext object, which will be
processed using the tls_set_context method.
Defaults to None, which indicates that TLS should not be used.
:param protocol: the MQTT protocol version to use. Defaults to MQTTv311.
:param transport: set to "tcp" to use the default setting of transport which is
raw TCP. Set to "websockets" to use WebSockets as the transport.
:param clean_session: a boolean that determines the client type. If True,
the broker will remove all information about this client
when it disconnects. If False, the client is a persistent
client and subscription information and queued messages
will be retained when the client disconnects.
Defaults to True. If protocol is MQTTv50, clean_session
is ignored.
:param proxy_args: a dictionary that will be given to the client.
"""
Tff7b72if Te6edf3msg_count Tff7b72< T79c0ff1Tb4b4b4:
Tff7b72raise Tf85149ValueErrorTb4b4b4(Ta5d6ff'Ta5d6ffmsg_count must be > 0Ta5d6ff'Tb4b4b4)
T8b949e# Set ourselves up to return a single message if msg_count == 1, or a list
T8b949e# if > 1.
Tff7b72if Te6edf3msg_count Tff7b72== T79c0ff1Tb4b4b4:
Te6edf3messages Tff7b72= Tff7b72None
Tff7b72elseTb4b4b4:
Te6edf3messages Tff7b72= Tb4b4b4[Tb4b4b4]
T8b949e# Ignore clean_session if protocol is MQTTv50, otherwise Client will raise
Tff7b72if Te6edf3protocol Tff7b72== Te6edf3pahoTff7b72.Td2a8ffMQTTv5Tb4b4b4:
Te6edf3clean_session Tff7b72= Tff7b72None
Te6edf3userdata Tff7b72= Tb4b4b4{Ta5d6ff'Ta5d6ffretainedTa5d6ff'Tb4b4b4:Te6edf3retainedTb4b4b4, Ta5d6ff'Ta5d6ffmsg_countTa5d6ff'Tb4b4b4:Te6edf3msg_countTb4b4b4, Ta5d6ff'Ta5d6ffmessagesTa5d6ff'Tb4b4b4:Te6edf3messagesTb4b4b4}
Te6edf3callbackTb4b4b4(Te6edf3_on_message_simpleTb4b4b4, Te6edf3topicsTb4b4b4, Te6edf3qosTb4b4b4, Te6edf3userdataTb4b4b4, Te6edf3hostnameTb4b4b4, Te6edf3portTb4b4b4,
Te6edf3client_idTb4b4b4, Te6edf3keepaliveTb4b4b4, Te6edf3willTb4b4b4, Te6edf3authTb4b4b4, Te6edf3tlsTb4b4b4, Te6edf3protocolTb4b4b4, Te6edf3transportTb4b4b4,
Te6edf3clean_sessionTb4b4b4, Te6edf3proxy_argsTb4b4b4)
Tff7b72return Te6edf3userdataTb4b4b4[Ta5d6ff'Ta5d6ffmessagesTa5d6ff'Tb4b4b4]
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────